package com.xujianlong.app.func;

import com.alibaba.fastjson.JSONObject;
import com.xujianlong.common.GmallConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import com.xujianlong.utils.DimUtil;
import com.xujianlong.utils.ThreadPoolUtil;

import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimInfoJoinFunction<T> {
    private Connection connection;
    private ThreadPoolExecutor threadPoolExecutor;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        threadPoolExecutor= ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture)  {
            threadPoolExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        //查询维度表
                        JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, getKey(t));
                        //补充这个维度的信息
                        join(t,dimInfo);
                        //写出数据,要的是集合，单个数据的集合
                        resultFuture.complete(Collections.singletonList(t));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
    }





    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) {
        System.out.println("TimeOut:"+input);
    }
}
